package com.tzg157.fitness.kafka;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import org.apache.spark.api.java.Optional;
import scala.Tuple4;

import javax.annotation.PostConstruct;
import java.io.Serializable;
import java.util.*;

@Component
//@RequiredArgsConstructor
public class RealMsgReceiver implements Serializable {

    private static final String topic = "flumeRealLog";
    private static final String producerTopic = "venueRealMsg";
    private Map<String, Object> kafkaParams;

//    private final JavaStreamingContext localStreamingContext;
//
//    private final MsgResolver msgResolver;
//    private final SparkConf sparkConf;
//    private final SparkSession spark;

//    @PostConstruct
    public void init() {
        // Kafka 配置
        kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "localhost:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "user_real_group_id");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);

        System.setProperty("HADOOP_USER_NAME", "root");
    }

    public static void main(String[] args) throws InterruptedException {
        RealMsgReceiver receiver = new RealMsgReceiver();
        receiver.receive();
    }

    public void receive() throws InterruptedException {
        System.setProperty("HADOOP_USER_NAME", "root");
        SparkConf conf = new SparkConf().setAppName("fitnessReal").setMaster("local[4]");
        // 创建 JavaStreamingContext
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

        jssc.checkpoint("hdfs://localhost:8020/checkpoint/fitnessReal");
        // 创建 SparkSession
        SparkSession spark = SparkSession.builder()
                .config(jssc.sparkContext().getConf())
                .enableHiveSupport()
                .config("hive.metastore.uris", "thrift://localhost:9083")
                .getOrCreate();

        JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
        // 订阅的 Kafka 主题
        Collection<String> topics = Collections.singletonList(topic);
        kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "localhost:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "user_real_group_id");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);

        // 创建 Kafka DStream
        JavaInputDStream<ConsumerRecord<String, String>> stream =
                KafkaUtils.createDirectStream(
                        jssc,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.Subscribe(topics, kafkaParams)
                );
        // 处理接收到的数据
        JavaDStream<JSONObject> newDataStream = stream.flatMap(record -> {
            String content = record.value();
            if (StrUtil.isBlank(content)) {
                return Collections.emptyIterator();
            }
            JSONObject jsonObject = JSONUtil.parseObj(content);
            return Arrays.asList(jsonObject).iterator();

        });
        // 将数据转换为键值对 (venueId, 1 或 -1)
        JavaPairDStream<Long, Integer> venueDStream = newDataStream.flatMapToPair((PairFlatMapFunction<JSONObject, Long, Integer>) jsonObject -> {
            String type = jsonObject.getStr("type");
            JSONObject data = jsonObject.getJSONObject("data");
            if (data != null) {
                Long venueId = data.getLong("venueId");
                if (venueId != null) {
                    int count = "venueSignIn".equalsIgnoreCase(type) ? 1 : -1;
                    return Arrays.asList(new Tuple2<>(venueId, count)).iterator();
                }
            }
            return Collections.emptyIterator();
        });

        // 定义状态更新函数
        Function2<List<Integer>, Optional<Tuple2<Integer, Boolean>>, Optional<Tuple2<Integer, Boolean>>> updateFunction =
                (values, state) -> {
                    int sum = values.stream().mapToInt(Integer::intValue).sum();
                    if (state.isPresent()) {
                        int newState = state.get()._1() + sum;
                        return Optional.of(new Tuple2<>(newState, newState != state.get()._1()));
                    } else {
                        return Optional.of(new Tuple2<>(sum, true));
                    }
                };

        // 使用 updateStateByKey 来更新每个教室的人数，并标记是否有变化
        JavaPairDStream<Long, Tuple2<Integer, Boolean>> venueCounts = venueDStream.updateStateByKey(updateFunction);


        // 过滤出有变化的数据
        JavaPairDStream<Long, Integer> changedVenueCounts = venueCounts.filter(record -> record._2()._2())
                .mapToPair(record -> new Tuple2<>(record._1(), record._2()._1()));

        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 将 venueCounts 发送到 Kafka
        changedVenueCounts.foreachRDD(rdd -> {
            rdd.foreachPartition(partitionOfRecords -> {
                while (partitionOfRecords.hasNext()) {
                    Tuple2<Long, Integer> record = partitionOfRecords.next();
                    String key = record._1().toString();
                    String value = record._2().toString();
                    Map<String,Object> map = new HashMap<>();
                    map.put("venueId",record._1());
                    map.put("num",record._2());
                    ProducerRecord<String, String> producerRecord = new ProducerRecord<>(producerTopic, key, JSONUtil.toJsonStr(map));
                    KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
                    producer.send(producerRecord);
                }
            });
        });
        venueCounts.print();
        // 启动 StreamingContext
        jssc.start();
        jssc.awaitTermination();
    }
}
